package common

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable
import scala.collection.mutable.ListBuffer
import scala.util.parsing.json.JSON

object Tools {
  //将字段间用(<>)分割,字段的key和值之间用\t分割的记录转化成json串
  def get_json_file(lines: RDD[String]): RDD[String] = {
    var json_line = ""
    val json_line_result: RDD[String] = lines.map { line => {
      json_line = "{"
      val ks_vs: Array[String] = line.split("\\(<>\\)")
      for (k_v <- ks_vs) {
        val key_value: Array[String] = k_v.split("\t")
        if (key_value.length == 2)
          json_line = json_line + "\"" + key_value(0) + "\"" + ":" + "\"" + key_value(1) + "\","
        else if (key_value.length == 1)
          json_line = json_line + "\"" + key_value(0) + "\"" + ":" + "\"\","
      }
      json_line = json_line.dropRight(1) + "}"
    }
      json_line
    }
    json_line_result
  }

  //将分布在多行的json一条json串,转换成一行
  def AJsonlineInMutiLineTransToOneLine(lines:RDD[String]):RDD[String]={
    val split_lines: RDD[String] = lines.flatMap(line => line.split("}")).filter(line=>line!="")
    val res_lines:RDD[String]=split_lines.map(line=>{
      line+"}"
    })
    res_lines
  }


  //处理json串为如下格式
  //字段间用(<>)分割,字段的key和值之间用\t分割
  def jsonFormat(json_file: RDD[String]): RDD[String] = {
    val records: RDD[Option[Any]] = json_file.map(one_line => JSON.parseFull(one_line))
    val result_maps: RDD[String] = records.map {
      val myMap: mutable.HashMap[String, String] = new mutable.HashMap[String, String]()
      r =>
        r match { //r:Option[Any]
          case Some(maps: Map[String, Any]) => { //maps:Map[String,Any]
            maps.foreach(x => myMap += (x._1 -> x._2.toString))
          }
        }
        myMap.toMap.mkString("(<>)").replace("->", "\t")
    }
    result_maps
  }

  //将json串转化成REC格式
  //  <REC>
  //  <大写字段名1>=值1
  //  <大写字段名2>=值2
  def jsonFormatREC(json_file: RDD[String]): RDD[String] = {
    val records: RDD[Option[Any]] = json_file.map(one_line => JSON.parseFull(one_line))
    val result_maps: RDD[String] = records.map {
      val myMap: mutable.HashMap[String, String] = new mutable.HashMap[String, String]()
      r =>
        r match { //r:Option[Any]
          case Some(maps: Map[String, Any]) => { //maps:Map[String,Any]
            maps.foreach(x => myMap += (x._1 -> x._2.toString))
          }
        }
        var str = "<REC>"
        val keys: List[String] = myMap.keys.toList
        keys.map { key => {
          str += "\n<" + key.toUpperCase + ">=" + myMap.getOrElse(key, "")
        }
        }
        str
    }
    result_maps
  }
  //  //将REC文件转换成json串(少量数据,对每行进行处理)
  //  def RECFormatJson(String):String={
  //    val arr=new Array[String]()
  //  }


  def jsonFormatRECToCSYD_TABLEMETA_WORLD(json_file: RDD[String]): RDD[String] = {
    val records: RDD[Option[Any]] = json_file.map(one_line => JSON.parseFull(one_line))
    val result_maps: RDD[String] = records.map {
      val myMap: mutable.HashMap[String, String] = new mutable.HashMap[String, String]()
      r =>
        r match { //r:Option[Any]
          case Some(maps: Map[String, Any]) => { //maps:Map[String,Any]
            maps.foreach(x => myMap += (x._1 -> x._2.toString))

            val keys = myMap.keys.toList
            keys.map { key => {
              if (key == "tm_time") {
                myMap += ("TM_时间" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "tm_address") {
                myMap += ("TM_地区" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "complete_indicator") {
                myMap += ("TM_完整指标" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "tm_original_value") {
                myMap += ("TM_原始数值" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "tm_original_uint") {
                myMap += ("TM_原始单位" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "tm_common_value") {
                myMap += ("TM_标准数值" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "tm_common_uint") {
                myMap += ("TM_标准单位" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "address_code") {
                myMap += ("地域代码" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "tm_fuzzy_address") {
                myMap += ("TM_模糊地区" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "knowledge_element") {
                myMap += ("TM_知识元" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "address_level") {
                var tmp = myMap.getOrElse(key, "")
                if (!tmp.isEmpty) tmp = tmp.toDouble.toInt.toString
                myMap += ("TM_地区等级" -> tmp)
                myMap -= (key)
              }
              if (key == "inbound_identity") {
                var tmp = myMap.getOrElse(key, "")
                if (!tmp.isEmpty) tmp = tmp.toDouble.toInt.toString
                myMap += ("入库标识" -> tmp)
                myMap -= (key)
              }
              if (key == "excel_year_code") {
                myMap += ("EXCEL年鉴代码" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "year_code_chinese_name") {
                myMap += ("年鉴中文名" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "year_year") {
                myMap += ("年鉴年份" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "entry_title") {
                myMap += ("条目题名" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "quote") {
                myMap += ("引文" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "gauncangbanzhuanjizhuantidaima") {
                myMap += ("馆藏版专辑专题代码" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "year") {
                myMap += ("年份" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "address_show") {
                myMap += ("地区显示" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "tm_province") {
                myMap += ("TM_省" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "valid_tag") {
                var tmp = myMap.getOrElse(key, "")
                if (!tmp.isEmpty) tmp = tmp.toDouble.toInt.toString
                myMap += ("VALIDTAG" -> tmp)
                myMap -= (key)
              }
              if (key == "rearch_indicator") {
                myMap += ("检索指标" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "tm_fuzzy_indicator") {
                myMap += ("TM_模糊指标" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "tm_knowledge_element") {
                myMap += ("TM_知识元S" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
            }
            }
            var str = "<REC>"
            val new_keys = myMap.keys.toList
            new_keys.map { key => {
              str += "\n<" + key + ">=" + myMap.getOrElse(key, "")
            }
            }
            str
          } //some
        } //match
    } //map
    result_maps
  }

  def jsonFormatRECToYEARBOOKTABLEDATA2008_FORMAL_GUOJI(json_file: RDD[String]): RDD[String] = {
    val records: RDD[Option[Any]] = json_file.map(one_line => JSON.parseFull(one_line))
    val result_maps: RDD[String] = records.map {
      val myMap: mutable.HashMap[String, String] = new mutable.HashMap[String, String]()
      r =>
        r match { //r:Option[Any]
          case Some(maps: Map[String, Any]) => { //maps:Map[String,Any]
            maps.foreach(x => myMap += (x._1 -> x._2.toString))

            val keys = myMap.keys.toList
            keys.map { key => {
              if (key == "my_year") {
                myMap += ("年份" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "address") {
                myMap += ("地区" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "format_indicator") {
                myMap += ("正式指标" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "format_value") {
                myMap += ("正式数值" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "format_unit") {
                myMap += ("正式单位" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "original_value") {
                myMap += ("原始数值" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "original_unit") {
                myMap += ("原始单位" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "knowledge_element") {
                myMap += ("知识元" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "content_notes") {
                myMap += ("内容注释" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "year_chinese") {
                myMap += ("年鉴中文名" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "year_year") {
                myMap += ("年鉴年份" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "address_code") {
                myMap += ("地区代码" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "address_level") {
                var tmp = myMap.getOrElse(key, "")
                if (!tmp.isEmpty) tmp = tmp.toDouble.toInt.toString
                myMap += ("地区等级" -> tmp)
                myMap -= (key)
              }
              if (key == "fuzzy_address") {
                myMap += ("模糊地区" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "search_indicator") {
                myMap += ("检索指标" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "fuzzy_indicator") {
                myMap += ("模糊指标" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "inbound_indentify") {
                var tmp = myMap.getOrElse(key, "")
                if (!tmp.isEmpty) tmp = tmp.toDouble.toInt.toString
                myMap += ("入库标识" -> tmp)
                myMap -= (key)
              }
              if (key == "year_code") {
                myMap += ("EXCEL年鉴代码" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "valid_tag") {
                var tmp = myMap.getOrElse(key, "")
                if (!tmp.isEmpty) tmp = tmp.toDouble.toInt.toString
                myMap += ("VALIDTAG" -> tmp)
                myMap -= (key)
              }
              if (key == "year_name") {
                myMap += ("年鉴名称" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "indicator_type") {
                myMap += ("指标类别" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "indicator_sort") {
                var tmp = myMap.getOrElse(key, "")
                if (!tmp.isEmpty) tmp = tmp.toDouble.toInt.toString
                myMap += ("指标排序" -> tmp)
                myMap -= (key)
              }
              if (key == "belongs_continent") {
                myMap += ("地域类型代码" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "file_name") {
                myMap += ("文件名" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "item_name") {
                myMap += ("条目题名" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "tm_fuzzy_indicator") {
                myMap += ("TM_模糊指标" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "show_value") {
                myMap += ("显示数值" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "show_unit") {
                myMap += ("显示单位" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "file_name_x") {
                myMap += ("文件名X" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "year_chinese_name") {
                myMap += ("年鉴中文名X" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "year_year_x") {
                myMap += ("年鉴年份X" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "item_title_x") {
                myMap += ("条目题名X" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "code_x") {
                myMap += ("编号X" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "inbound_indentify_x") {
                var tmp = myMap.getOrElse(key, "")
                if (!tmp.isEmpty) tmp = tmp.toDouble.toInt.toString
                myMap += ("入库标识X" -> tmp)
                myMap -= (key)
              }
              if (key == "country_flag_address_code") {
                myMap += ("国标地域代码" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }

            }
            }
            var str = "<REC>"
            val new_keys = myMap.keys.toList
            new_keys.map { key => {
              str += "\n<" + key + ">=" + myMap.getOrElse(key, "")
            }
            }
            str
          } //some
        } //match
    } //map
    result_maps
  }

  def jsonFormatEARBOOKTABLEDATA2008_FORMAL_REPORTGROUPER_GUOJI_2019_ZL(json_file: RDD[String]): RDD[String] = {
    val records: RDD[Option[Any]] = json_file.map(one_line => JSON.parseFull(one_line))
    val result_maps: RDD[String] = records.map {
      val myMap: mutable.HashMap[String, String] = new mutable.HashMap[String, String]()
      r =>
        r match { //r:Option[Any]
          case Some(maps: Map[String, Any]) => { //maps:Map[String,Any]
            maps.foreach(x => myMap += (x._1 -> x._2.toString))
            val keys = myMap.keys.toList
            keys.map { key => {
              if (key == "year") {
                myMap += ("年份" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "formal_indicator") {
                myMap += ("正式指标" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "reporter_code") {
                myMap += ("地区代码" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "address_level") {
                var tmp = myMap.getOrElse(key, "")
                if (!tmp.isEmpty) tmp = tmp.toDouble.toInt.toString
                myMap += ("地区等级" -> tmp)
                myMap -= (key)
              }
              if (key == "report_grouper") {
                myMap += ("REPORTGROUPER" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "show_value") {
                myMap += ("显示数值" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "show_unit") {
                myMap += ("显示单位" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "file_name_x") {
                myMap += ("文件名X" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "year_chinese_name") {
                myMap += ("年鉴中文名X" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "year_year_x") {
                myMap += ("年鉴年份X" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "item_head_name_x") {
                myMap += ("条目题名X" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "num_x") {
                myMap += ("编号X" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "inbound_flag") {
                var tmp = myMap.getOrElse(key, "")
                if (!tmp.isEmpty) tmp = tmp.toDouble.toInt.toString
                myMap += ("入库标识X" -> tmp)
                myMap -= (key)
              }
              if (key == "belongs_continent") {
                myMap += ("地域类型代码" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
              if (key == "internationnal_place_num") {
                myMap += ("国标地域代码" -> myMap.getOrElse(key, ""))
                myMap -= (key)
              }
            }
            }
            var str = "<REC>"
            val new_keys = myMap.keys.toList
            new_keys.map { key => {
              str += "\n<" + key + ">=" + myMap.getOrElse(key, "")
            }
            }
            str
          }
        }
    }
    result_maps
  }


  def main(args: Array[String]): Unit = {
    val sc: SparkContext = getSC("NAVIFREQ_ALL_CLASSIFY_WORLDWEB_Tools")
    ifDirExistsDelete("/tmp/InternationalData_Result/YEARBOOKTABLEDATA2008_FORMAL_REPORTGROUPER_GUOJI_2019_ZL/")

  }


  //该方法可以得到指定hdfs目录下的所有文件全路径名
  //参数hdfsDir必须以"/"结束
  def getHDFSDirFiles(hdfsDir: String): Array[String] = {
    val conf = new Configuration()
    conf.set("fs.defaultFS", "hdfs://node01:8020")
    val fileSystem: FileSystem = FileSystem.get(conf)
    val dirPath = new Path(hdfsDir)
    val iterator = fileSystem.listFiles(dirPath, true)
    val list = new ListBuffer[String]
    while (iterator.hasNext) {
      val pathStatus = iterator.next
      val hdfsPath = pathStatus.getPath.toString
      list += hdfsPath
    }
    fileSystem.close()
    list -= ("hdfs://node01:8020" + hdfsDir + "_SUCCESS")
    list.toArray
  }

  //如果hdfs上存在该目录则删除
  def ifDirExistsDelete(hdfsDir: String): Unit = {
    val conf = new Configuration()
    conf.set("fs.defaultFS", "hdfs://node01:8020")
    val fileSystem = FileSystem.get(conf)
    // 判断HDFS中是否存在该目录，如果存在则删除
    if (fileSystem.exists(new Path(hdfsDir))) {
      // 第二个参数代表循环删除
      fileSystem.delete(new Path(hdfsDir), true)
    }
  }

  def getSparkConf(appName: String): SparkConf = {
    val sparkConf: SparkConf = new SparkConf().setAppName(appName).setMaster("local[*]") //集群运行去掉setMaster
    sparkConf
  }

  def getSC(appName: String) = {
    val sparkConf = getSparkConf(appName)
    val sc: SparkContext = new SparkContext(sparkConf)
    sc.setLogLevel("ERROR")
    sc
  }

  def getSpark(appName: String): SparkSession = {
    val sparkConf: SparkConf = getSparkConf(appName)
    val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    spark
  }

  //在数据倾斜的情况下，查看出现次数最多的10个key
  def sampleTopKey(sparkSession: SparkSession, tableName: String, keycolumn: String): Array[(Int, Row)] = {
    val df = sparkSession.sql("select" + keycolumn + "from" + tableName)
    val top10Key = df.select(keycolumn)
      .sample(false, 0.01).rdd //对key不放回采样
      .map(k => (k, 1)).reduceByKey(_ + _) //统计不同key出现的次数
      .map(k => (k._2, k._1)).sortByKey(false) //对统计key降序排序
      .take(10) //拿到出现次数最多的10个key
    top10Key
  }


}
